package org.bloodnoteii.cloudprovider.hello.services;

import com.alibaba.fastjson.JSONObject;
import org.bloodnoteii.cloudprovider.hello.mapper.mysql.HelloMapper;
import org.bloodnoteii.cloudprovider.hello.mapper.neo4j.HelloNeo4jMapper;
import org.bloodnoteii.cloudprovider.hello.model.HelloModel;
import org.bloodnoteii.cloudprovider.hello.model.Neo4jModel;
import org.bloodnoteii.cloudprovider.utils.RedisUtils;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Service
public class HelloServices {
    private static final Logger log = LoggerFactory.getLogger(HelloServices.class);

    @Value("${spring.kafka.topic:test_01}")
    private String topic;

    @Autowired
    private HelloMapper helloMapper;

    @Autowired
    private HelloNeo4jMapper helloNeo4jMapper;

    @Autowired
    private RestHighLevelClient restClient;

    @Autowired
    private RedisUtils redisUtils;

    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    public JSONObject helloProvider(Integer id) {
        JSONObject JS = new JSONObject();
        List<HelloModel> result = helloMapper.getModelBase(id);
        List<Neo4jModel> res = helloNeo4jMapper.getNeo4jBase(id);
        log.info("mybatis");
        log.info(res.toString());
        JS.put("mysqlValue", result);
        JS.put("neo4jValue", res);
        String kvalue = JSONObject.toJSONString(redisUtils.get("MTMwNEMyQkVFMkZFRDQ0Qw=="));
        JS.put("redis", kvalue);
        SearchRequest searchRequest = new SearchRequest("tablecustomer");
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchPhraseQuery("title", "菜菜"));
        sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));

        searchRequest.source(sourceBuilder);
        try {
            log.info("ES查询语句-------------------------");
            log.info(searchRequest.toString());
            SearchResponse response = restClient.search(searchRequest, RequestOptions.DEFAULT);
            Arrays.stream(response.getHits().getHits()).forEach(i -> {
                log.info(i.getIndex());
                log.info("es返回数据：{}", i.getSourceAsMap());
                JS.put("esValue", i.getSourceAsMap());
            });
            log.info(response.getHits().getTotalHits().toString());
        } catch (IOException e) {
            e.printStackTrace();
        }

        kafkaTemplate.send(topic, JS.toJSONString());
        return JS;
    }
}
